package com.pw.study.flink.sql;

import com.pw.study.flink.entities.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.lit;

/**
 * @Author: linux_future
 * @since: 2022/3/11
 **/
public class $10SqlWindow {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        mc1(env);

    }

    private static void mc1(StreamExecutionEnvironment env) {
        SingleOutputStreamOperator<WaterSensor> data = env.fromElements(
                new WaterSensor("sensor_1", 1000L, 10),
                new WaterSensor("sensor_1", 2000L, 20),
                new WaterSensor("sensor_2", 3000L, 30),
                new WaterSensor("sensor_1", 4001L, 40),
                new WaterSensor("sensor_1", 5000L, 50),
                new WaterSensor("sensor_2", 6000L, 60)
        ).assignTimestampsAndWatermarks(WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner((wc, ts) -> wc.getTs()));
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        Table table = tEnv.fromDataStream(data, $("id"), $("ts").rowtime(), $("vc"));

        wc2(tEnv, table);
        wc3(tEnv, table);
        wc4(tEnv, table);
        wc1(table);

    }

    private static void wc4(StreamTableEnvironment tEnv, Table table) {
        System.out.println("session Window2...........................");
        tEnv.createTemporaryView("sensor4", table);
        tEnv
                .sqlQuery("select " +
                        "   id, " +
                        "   session_start(ts, interval '2' second) stt, " +
                        "   session_end(ts, interval '2' second) edt, " +
                        "   sum(vc) vc_sum " +
                        "from sensor4 " +
                        "group by id, session(ts, interval '2' second)")
                .execute()
                .print();

    }

    private static void wc3(StreamTableEnvironment tEnv, Table table) {
        System.out.println("slide Window...........................");
        tEnv.createTemporaryView("sensor3", table);
        tEnv
                .sqlQuery("select " +
                        "   id, " +
                        "   hop_start(ts, interval '2' second,interval '5' second) stt, " +
                        "   hop_end(ts, interval '2' second,interval '5' second) edt, " +
                        "   sum(vc) vc_sum " +
                        "from sensor3 " +
                        "group by id, hop(ts, interval '2' second,interval '5' second)")
                .execute()
                .print();
    }

    private static void wc2(StreamTableEnvironment tEnv, Table table) {
        System.out.println("tumble window2..............................");
        tEnv.createTemporaryView("sensor1", table);
        tEnv
                .sqlQuery("select " +
                        "   id, " +
                        "   tumble_start(ts, interval '5' second) stt, " +
                        "   tumble_end(ts, interval '5' second) edt, " +
                        "   sum(vc) vc_sum " +
                        "from sensor1 " +
                        "group by id, tumble(ts, interval '5' second)")
                .execute()
                .print();
    }

    private static void wc1(Table table) {
        System.out.println("window................................");
        GroupWindow w = Tumble.over(lit(5).second()).on($("ts")).as("win");
        GroupWindow w1 = Slide.over(lit(5).second()).every(lit(2).second()).on("ts").as("win");
        GroupWindow w2 = Session.withGap(lit(2).second()).on("ts").as("win");
        table.window(w2).groupBy($("id"), $("win"))
                .select(
                        $("id"), $("win").start().as("st"),
                        $("win").end().as("end"),
                        $("vc").sum().as("vc_sum")
                ).execute().print();

    }

}
